#include <errno.h>
#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/eventfd.h>
#include <sys/epoll.h>
#include <utils.h>
#include <dlist.h>
#include <evtloop.h>
#include <async.h>
#include <printk.h>
#include <coroutines.h>

static int ccb_exist_nolock(ev_loop_t *loop, ccb_t *ccb)
{
    ccb_t *iter;

    list_for_each_entry(iter, &loop->cq, entry)
    {
        if (iter == ccb)
            return 1;
    }
    return 0;
}
static int ccb_exist(ev_loop_t *loop, ccb_t *ccb)
{
    int r;

    svc_mutex_lock(&loop->cq_mutex);
    r = ccb_exist_nolock(loop, ccb);
    svc_mutex_unlock(&loop->cq_mutex);
    return r;
}

int ccb_add(ev_loop_t *loop, ccb_t *ccb, void (*cb)(ccb_t *ccb), void *data)
{
    assert(ev__is_active(loop));

    if (ccb_exist(loop, ccb))
        return -EEXIST;

    ccb->loop = loop;
    ccb->cb   = cb;
    ccb->data = data;
    ccb->pending = 0;
    memset(&ccb->_pt, 0, sizeof(ccb->_pt));
    INIT_KFIFO(ccb->msgqueue);

    svc_mutex_lock(&loop->cq_mutex);
    list_add_tail(&ccb->entry, &loop->cq);
    svc_mutex_unlock(&loop->cq_mutex);

    ev__req_add(loop, ccb);
    svc_async_send(&loop->cq_async);
    return 0;
}
void ccb_del(ccb_t *ccb)
{
    ev_loop_t *loop = ccb->loop;

    assert(ev__is_active(loop));
    if (!ccb_exist(loop, ccb))
        return;

    svc_mutex_lock(&loop->cq_mutex);
    list_del(&ccb->entry);
    svc_mutex_unlock(&loop->cq_mutex);

    ev__req_del(loop, ccb);
}

int ccb_sig_kill(ccb_t *ccb, ccb_sig_t sig)
{
    ev_loop_t *loop = ccb->loop;

    assert(ev__is_active(loop));
    svc_mutex_lock(&loop->cq_mutex);
    ccb->pending |= sig;
    svc_mutex_unlock(&loop->cq_mutex);
    svc_async_send(&loop->cq_async);
    return 0;
}
ccb_sig_t ccb_sig_get(ccb_t *ccb)
{
    return ccb->pending;
}

int ccb_msg_push(ccb_t *ccb, ccb_msg_t *msg)
{
    int r;
    ev_loop_t *loop = ccb->loop;

    assert(ev__is_active(loop));
    r = kfifo_put(&ccb->msgqueue, *msg);
    if (r > 0)
        svc_async_send(&loop->cq_async);
    return r;
}

int ccb_msg_pop(ccb_t *ccb, ccb_msg_t *msg)
{
    return kfifo_get(&ccb->msgqueue, msg);
}

void ccb_schedule(svc_async_t *handle)
{
    ev_loop_t *loop = container_of(handle, ev_loop_t, cq_async);
    ccb_t *ccb, *n;

    svc_mutex_lock(&loop->cq_mutex);
    list_for_each_entry_safe(ccb, n, &loop->cq, entry)
    {
        /* any ccb may be deleted during execution.
         */
        if (!ccb_exist_nolock(loop, ccb))
        {
            n = list_first_entry(&loop->cq, ccb_t, entry);
            continue;
        }
        svc_mutex_unlock(&loop->cq_mutex);
        ccb->cb(ccb);
        if (ccb_status(ccb) == PT_STATUS_FINISHED)
            ccb_del(ccb);
        svc_mutex_lock(&loop->cq_mutex);
    }
    svc_mutex_unlock(&loop->cq_mutex);
}
